package com.atguigu.gmall.utils

import java.util.Objects
import java.util

import io.searchbox.client.config.HttpClientConfig
import io.searchbox.client.{JestClient, JestClientFactory}
import io.searchbox.core.{Bulk, BulkResult, Index}

/**
  * @ClassName: MyESuTIL
  * @Description:
  * @Author: kele
  * @Date: 2021/3/25 14:04
  **/


object MyESUtil {

  private val ES_HOST = "http://hadoop102"
  private val ES_HTTP_PORT = 9200
  private var factory:JestClientFactory = null

  /**
    * 获取客户端
    *
    * @return jestclient
    */
  def getClient: JestClient = {
    if (factory == null) build()
    factory.getObject
  }

  /**
    * 关闭客户端
    */
  def close(client: JestClient): Unit = {
    if (!Objects.isNull(client)) try
      client.shutdownClient()
    catch {
      case e: Exception =>
        e.printStackTrace()
    }
  }

  /**
    * 建立连接
    */
  private def build(): Unit = {
    factory = new JestClientFactory
    factory.setHttpClientConfig(new HttpClientConfig.Builder(ES_HOST + ":" + ES_HTTP_PORT)
      .multiThreaded(true)
      .maxTotalConnection(50) //连接总数
      .connTimeout(10000).readTimeout(10000)
      .build)
  }

  /*
      批量插入数据到ES
          indexName: 写入的index名
          docList：  写入的数据，封装为一个List集合
   */
  def  insertBulk(indexName:String,docList:List[(String,Any)]): Unit ={

    if(docList.size>0){
      val jest: JestClient = getClient
      val bulkBuilder = new Bulk.Builder().defaultIndex(indexName).defaultType("_doc")

      for ((id,doc) <- docList ) {
        val indexBuilder = new Index.Builder(doc)
        if(id!=null){
          indexBuilder.id(id)
        }
        val index: Index = indexBuilder.build()
        bulkBuilder.addAction(index)
      }
      val bulk: Bulk = bulkBuilder.build()

      var items: util.List[BulkResult#BulkResultItem] = null
      try {
        items = jest.execute(bulkBuilder.build()).getItems
      } catch {
        case ex: Exception => println(ex.toString)
      } finally {
        close(jest)
        println("保存" + items.size() + "条数据")
        // <- 只能用于遍历scala的集合，目前items是java的集合，将java集合转 scala集合  java集合.asScala
        // scala转java  scala集合.asJava

        //将java中的集合转化为scala中的集合
        import collection.JavaConverters._
        for (item <- items.asScala) {
          if (item.error != null && item.error.nonEmpty) {
            println(item.error)
            println(item.errorReason)
          }
        }
      }
    }
  }


}
